package com.itheima.kafka.stream;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.apache.kafka.streams.kstream.Windowed;

import java.util.Arrays;
import java.util.Properties;

/**
 * @Description:  接收到数据 计算单词出现的次数
 * @Version: V1.0
 * 如：接收消息—— 数据源
 *
 * * hello kafka streams
 * * hello heima kafka
 * * hello shanghai heima kafka
 *
 * 结果输出：
 *
 *  * hello： 3
 *  * kafka： 3
 *  * streams：1
 *  * heima： 2
 *  * shanghai： 1
 */
public class KafkaStreamFastStart {


    public static void main(String[] args) {

        //1 kafka配置信息
        Properties prop = new Properties();
        prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.129:9092");
        prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        prop.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-sample");

        //2 构建 计算拓扑
        StreamsBuilder builder = new StreamsBuilder();

        // 处理器
        streamProcessor(builder);

        // 创建 Kafka Stream 实时计算
        KafkaStreams streams = new KafkaStreams(builder.build(), prop);

        // 开启流实时计算
        streams.start();
    }

    // 处理数据
    private static void streamProcessor(StreamsBuilder builder) {

        // 接收数据
        KStream<String, String> stream = builder.stream("itcast-topic-input");

        // 计算  hello kafka streams  hello heima kafka
        stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
            @Override
            public Iterable<String> apply(String value) {
                System.out.println("接收到的待处理的消息：" + value);
                return Arrays.asList( value.split(" ") );
            }
        })

                .groupBy(new KeyValueMapper<String, String, String>() {
                    @Override
                    public String apply(String key, String value) {
                        System.out.println("k: "+key+" v: "+value);
                        return value;
                    }
                })

                .windowedBy(TimeWindows.of(5000))
                .count()
                // {key: "hello", value: "2"}
                .toStream()
                .map(new KeyValueMapper<Windowed<String>, Long, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> krWindowed, Long aLong) {
                        System.out.println("处理成功后的数据：" + krWindowed.key().toString());
                        System.out.println("处理成功后的数据：" + aLong.toString());
                        return new KeyValue<>(krWindowed.key().toString(), aLong.toString());
                    }
                })
                .to("itcast-topic-output");



    }


}
